package dht.socket;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.LinkedList;
import java.util.Queue;

import dht.entity.Message;
import dht.entity.NodeInfo;
import dht.throwable.DhtException;

public class SocketHandler implements Runnable {
	protected NodeInfo localInfo;
	protected Socket socket;
	protected ObjectOutputStream os;
	protected ObjectInputStream is;
	protected Queue<Message> msgQueue;

	public SocketHandler(NodeInfo localInfo, Socket socket) {
		this.localInfo = localInfo;
		this.socket = socket;
		this.msgQueue = new LinkedList<Message>();
		try {
			os = new ObjectOutputStream(this.socket.getOutputStream());
			is = new ObjectInputStream(this.socket.getInputStream());
		} catch (IOException e) {
			e.printStackTrace();
		}
		new Thread(this).start();
		System.out.println("starting socket: " + getSocketId());
	}

	@Override
	public void run() {
		Message msg = null;
		while (true) {
			synchronized (is) {
				try {
					msg = (Message) is.readObject();
				} catch (ClassNotFoundException e) {
					e.printStackTrace();
				} catch (IOException e) {
					// pipe broke, ignore
					break;
				}
				// System.out.println("receiving message... type: "
				// + msg.getMsgType() + " id: " + msg.getTransactionId()
				// + " from: " + msg.getSenderNodeInfo().getServerId()
				// + " size: " + msg.getMessageSize());
				MessageDelayer delayer = new MessageDelayer(localInfo, msg,
						msgQueue);
				new Thread(delayer).start();
			}
		}

	}

	public Message readMessage() throws DhtException {
		Message msg = null;
		synchronized (msgQueue) {
			if (!msgQueue.isEmpty()) {
				msg = msgQueue.poll();
			} else {
				try {
					msgQueue.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				msg = msgQueue.poll();
			}
		}
		System.out.println("receiving message... type: " + msg.getMsgType()
				+ " id: " + msg.getTransactionId() + " from: "
				+ msg.getSenderNodeInfo().getServerId() + " size: "
				+ msg.getMessageSize());
		return msg;
	}

	// public Message readMessage() throws DhtException {
	// Message msg = null;
	// try {
	// synchronized (is) {
	// msg = (Message) is.readObject();
	// System.out.println("receiving message... type: "
	// + msg.getMsgType() + " id: " + msg.getTransactionId()
	// + " from: " + msg.getSenderNodeInfo().getServerId()
	// + " size: " + msg.getMessageSize());
	// }
	// } catch (ClassNotFoundException e) {
	// e.printStackTrace();
	// throw new DhtException(e.getMessage(), e);
	// } catch (IOException e) {
	// e.printStackTrace();
	// throw new DhtException(e.getMessage(), e);
	// }
	// try {
	// Thread.sleep((long) (NodeInfo.cost(msg.getSenderNodeInfo(),
	// localInfo, msg.getMessageSize()) * 1000));
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return msg;
	// }

	public void sendMessage(Message msg) throws DhtException {
		// msg.setSenderNodeInfo(localInfo);
		try {
			synchronized (os) {
				System.out.println("sending message... type: "
						+ msg.getMsgType() + " id: " + msg.getTransactionId()
						+ " to: " + getSocketId() + " size: "
						+ msg.getMessageSize());
				os.writeObject(msg);
				os.flush();
			}
		} catch (IOException e) {
			throw new DhtException(e.getMessage(), e);
		}
	}

	public void closeSocket() throws DhtException {
		if (!socket.isClosed()) {
			System.out.println("closing socket: " + getSocketId());
			try {
				socket.close();
			} catch (IOException e) {
				e.printStackTrace();
				throw new DhtException(e.getMessage(), e);
			}
		}
	}

	protected String getRemoteIp() {
		return socket.getInetAddress().getHostAddress();
	}

	protected int getRemotePort() {
		return socket.getPort();
	}

	public String getSocketId() {
		return NodeInfo.getServerId(getRemoteIp(), getRemotePort());
	}

	@Override
	public boolean equals(Object obj) {
		SocketHandler socketHandler = (SocketHandler) obj;
		return getSocketId().equals(socketHandler.getSocketId());
	}

	@Override
	public int hashCode() {
		return getSocketId().hashCode();
	}

}